from __future__ import with_statement

import sys
import threading
try:
    import cPickle as pickle
except ImportError: #pragma: no cover
    import pickle
import kineta.berkdb as berkdb
import kineta.query as query
import kineta.threads as threads
import kineta.limits as limits
import kineta.utils as utils
from kineta.cmp import idx_key_factory
from kineta.error import KinetaError


class FileDB(utils.DeleteHelper):
    """
    Represents a connection to a database stored on a local filesystem.

    Instances of this class are thread-safe.  However, users should never
    open more than one FileDB object per database at the same time, even from
    separate processes.  If you wish to access a Kineta database from more than
    one process, host a L{Server} instance in a WSGI container, and use the
    L{RemoteDB} class to access it.
    """

    def _pre_init(self):
        self._db = None
        self._bgthread = None
        self._lockthread = None

    def __init__(self, path, cachesize="1MB"):
        """
        Create a new database or open an existing database.

        @type path: string
        @param path: The directory to create or open the database.  Should be
                     an absolute path, and must be an existing directory, not
                     a regular file.
        @type cachesize: string
        @param cachesize: The size of the database cache.  Can be specified as
                          bytes (B), kilobytes (KB), megabytes (MB) or
                          gigabytes (GB), using the appropriate suffix (no
                          suffix or a number instead of a string means bytes).
                          Fractional numbers are OK (e.g. 2.5GB).
                          In general, the cache should be equal to as much
                          memory as you can spare, because larger caches
                          generally lead to better performance.
        """
        cachebytes = utils.cache_bytes(cachesize)
        self.limits = limits.Limits(cachebytes)
        self._db = berkdb.BDB(path, idx_key_factory, self.limits, cachebytes)
        self._bgthread = threads.BGThread()
        self._lockthread = threads.LockDetectThread(self._db)

    def begin(self, isolated=False, txn=None):
        """
        Begin a transaction, which allows commiting or aborting a set of
        changes as an atomic unit.  To ensure transactions are always ended
        cleanly, this method should always be called as the initializer of a
        with: block.

        A transaction commits itself by default when control leaves the with:
        block it was created in, unless the block was terminated by raising an
        exception.  If the exception raised was of the type L{Rollback}, it
        will not be propagated up the stack.

        Transaction objects also have two methods which take no parameters,
        C{commit} and C{rollback}, in case calling C{begin} in a with: block
        is not possible.

        @type isolated: boolean
        @param isolated: If True, the transaction will be I{serializable},
                         meaning that data read by an earlier query will always
                         look the same to a later query.  This should not be
                         used unless it is neccessary as it reduces
                         concurrency.
        @type txn: transaction
        @param txn: A parent transaction.  Creates the new transaction as a
                    nested child of the parent, for multiple levels of
                    atomic granularity.
        @rtype: transaction
        @return: A new transaction object.
        """
        with _TxnLock(txn) as txnobj:
            return _Txn(self, self._db.txn_begin(isolated, txnobj))

    def close(self):
        """
        Close the database handle.  It is recommended to create FileDB
        instances using with: statements so that they will be always be closed
        after they have been used.  Otherwise, the caller should ensure this
        method is called to avoid database corruption.

        Calling this method after a database has already been closed has no
        effect.
        """
        if self._bgthread is not None:
            self._bgthread.quit()
            self._bgthread = None
        if self._lockthread is not None:
            self._lockthread.quit()
            self._lockthread = None
        if self._db is not None:
            self._db.close()
            self._db = None

    def __del__(self):
        """
        Calls L{close}.
        """
        self.close()

    def __enter__(self):
        """
        Provides support for creating database objects using the with:
        statement.
        """
        return self

    def __exit__(self, *exc_info):
        """
        Calls L{close} when a with: block is completed.
        """
        self.close()

    def set_bg_error_cb(self, callback):
        """
        Set a callback function which will be called when errors occur in a
        background task (such as creating an index, see also
        L{wait_for_bg_tasks}).  The default behavior with no callback
        set is to print the error to the console.

        Errors in background tasks are rare, however this method provides a
        way to change the default logging behavior in the event it is
        unacceptable.

        @type callback: callable
        @param callback: The error callback, should take one parameter which
                         will be an exception which was caught.
        """
        self._bgthread.set_error_cb(callback)

    def wait_for_bg_tasks(self):
        """
        Kineta performs maintanance tasks in a background thread to avoid
        blocking callers.  Calling this method will intentionally block
        the calling thread until any pending background tasks have completed.

        This may be useful, for example, to wait until an index has been
        created to run a query.
        """
        self._bgthread.queue_join()

    def insert(self, table, items, txn=None):
        """
        Insert items/rows (represented as dictionaries) into a table.

        @type table: string
        @param table: The table to insert into.  If the table does not exist,
                      it will be created.
        @type items: list of dicts
        @param items: A list of items a.k.a. rows, each one of which should be
                      a dict.  If a single dict is passed, it will be treated
                      as a list with one item. The keys of the dictionary are
                      treated as column names, and the values as values to
                      insert into the columns.  Values can be numbers,
                      strings, booleans, None, lists containing
                      any of these four types, or the special value
                      C{{"identity": True}} (a dict with the key "identity" set
                      to True) which inserts a unique sequence number for the
                      table, starting at 1.  If a single dict is passed, it
                      will be treated as a list with one item.
        @type txn: transaction
        @param txn: The transaction object to use.  If None, runs the method
                    outside the context of any transaction. Calling insert
                    on a transaction object itself will automatically call this
                    method with the txn parameter set.
        @rtype: list
        @return: A list of values, the length of which will be equal to the
                 length of items.  If an item which was inserted contained the
                 special identity value, the corresponding list entry will
                 contain the value which was inserted, otherwise it will
                 contain None.
        """
        with _TxnLock(txn) as txnobj:
            ret = []
            if not isinstance(items, (list, tuple)):
                items = [items]
            tableid = self._db.dict.get_id(table)
            with self._db.tables.lock() as lock:
                tblobj = self._db.tables.table_open(lock, tableid, txnobj,
                                                    False)
                seqobj = None
                try:
                    for item in items:
                        if None in item:
                            raise KinetaError("Cannot insert into a null "
                                              + "column name")
                        values = item.values()
                        hasidentity = FileDB._check_insert_values(values)
                        if hasidentity and seqobj is None:
                            seqobj = self._db.tables.sequence_open(lock,
                                                  tableid, txnobj)

                        ret.append(self._insert_item(lock, tblobj, seqobj,
                                                     item.keys(), values,
                                                     hasidentity, txnobj))
                finally:
                    if seqobj is not None:
                        seqobj.close()

            self._bgthread.task(lambda: self._db.tables.update_scores(len(ret),
                                                                      tableid))
            return ret

    def insert_bulk(self, table, columns, items, txn=None):
        """
        Insert items/rows (represented as lists with a single list of column
        names) into a table.

        @type table: string
        @param table: The table to insert into.  If the table does not exist,
                      it will be created
        @type columns: list of strings
        @param columns: A list of column names.  A single value will be treated
                       as a list with one item.
        @type items: list of lists
        @param items: A list of items a.k.a. rows, each one of which should be
                      a list of values.  The length of each sub-list must equal
                      the length of the columns list, and each item in the
                      sub-list will be inserted as a value in the corresponding
                      column with the same index.  Values can be numbers,
                      strings, booleans, None, lists containing
                      any of these four types, or the special value
                      C{{"identity": True}} (a dict with the key "identity" set
                      to True) which inserts a unique sequence number for the
                      table, starting at 1.  If a single dict is passed, it
                      will be treated as a list with one item.
        @type txn: transaction
        @param txn: The transaction object to use.  If None, calls the method
                    outside the context of any transaction. Calling
                    insert_bulk on a transaction object itself will
                    automatically call this method with the txn parameter set.
        @rtype: list
        @return: A list of values, the length of which will be equal to the
                 length of items.  If an item which was inserted contained the
                 special identity value, the corresponding list entry will
                 contain the value which was inserted, otherwise it will
                 contain None.
        """
        ret = []
        if not isinstance(columns, (list, tuple)):
            columns = [columns]
        if None in columns:
            raise KinetaError("Cannot insert into a null column name")
        with _TxnLock(txn) as txnobj:
            tableid = self._db.dict.get_id(table)
            with self._db.tables.lock() as lock:
                tblobj = self._db.tables.table_open(lock, tableid, txnobj,
                                                    False)
                seqobj = None
                try:
                    for item in items:
                        if len(columns) != len(item):
                            raise KinetaError("Length of columns and " +
                                           "values do not match (%s, %s)" %
                                           (len(columns), len(item)))
                        hasidentity = FileDB._check_insert_values(item)
                        if hasidentity and seqobj is None:
                            seqobj = self._db.tables.sequence_open(lock,
                                                  tableid, txnobj)

                        ret.append(self._insert_item(lock, tblobj, seqobj,
                                                     columns, item,
                                                     hasidentity, txnobj))
                finally:
                    if seqobj is not None:
                        seqobj.close()

            self._bgthread.task(lambda: self._db.tables.update_scores(len(ret),
                                                                      tableid))
            return ret

    def select(self, table, fields, where=None, joins=None, order=None,
               limit=None, offset=None, txn=None):
        """
        Query the database and return a row iterator.  Because it is important
        to close the row iterator when the query is complete, it is highly
        recommended to call this method as the initializer in a with: block.

        @type table: string or list
        @param table: The main table to query.  If this parameter is a list,
                      the first element is taken to be the table name, and the
                      second element a join name to use to distinguish the
                      main table name from any other joins (since all join
                      names must be unique, see the I{joins} parameter for
                      details).

                      Examples::

                        table="my_table"

                        table=["my_table", "my_alias"]
        @type fields: list
        @param fields: A list of columns to return for each row.  Items in the
                       list can either be a string denoting a column name to
                       return from the table specified by the I{table}
                       parameter, or a list with one item.  If the item is a
                       list, then the item inside the list must denote the name
                       of a join listed in the 'joins' parameter.  The
                       I{fields} in that join (if any) will then be inserted
                       into the result set in the place where the list is
                       present.  For example, the following would select two
                       columns ("c1" and "c2") from table "t1", and two columns
                       from table "t2" ("c3" and "c4"), for a total of four
                       columns for each row::

                          db.select(table="t1",
                                    fields=["c1", "c2", ["t2"]],
                                    joins=dict(table="t2", fields=["c3", "c4"],
                                               ...),
                                    ...)

        @type where: dict
        @param where: A dictionary in which each key/value pair specifies a
                      column and a restriction.  Rows must pass all the
                      restrictions in order to be returned by the query.  If
                      this parameter is None, all rows are returned.

                      Syntax::

                          where={
                              "column_1": param,
                              "column_2": [operator, param1, param2, ...],
                              ...
                          }

                      Parameters can be numbers, strings, booleans, None, or
                      lists containing any of these four types.  If a single
                      parameter is specified alone, e.g. C{"column": p}, it is
                      equivalent to C{"column": ["==", p]}.

                      The allowed operators are:

                          - C{==}, C{!=}, C{<}, C{>}, C{<=}, C{>=}
                            - Number of parameters: 1
                            - Passes if the column value is equal to, not equal
                              to, less than, greater than, less than or equal
                              to, or greater than or equal to the parameter,
                              respectively.
                          - C{in}
                            - Number of parameters: any
                            - Passes if the column value is equal to one or
                              more of the parameters.
                          - C{notin}
                            - Number of parameters: any
                            - Passes if the column value is not equal to any of
                              the parameters.
                          - C{between}
                            - Number of parameters: 2
                            - Passes if the column value is less than or equal
                              to the larger of the two arguments, and greater
                              than or equal to the smaller of the two
                              arguments.
                          - C{exclbetween}
                            - Number of parameters: 2
                            - Passes if the column value is less than the
                              larger of the two arguments, and greater than the
                              smaller of the two arguments.
                          - C{outside}
                            - Number of parameters: 2
                            - Passes if the column value is less than or equal
                              to the smaller of the two arguments, or greater
                              than or equal to the larger of the two arguments.
                          - C{excloutside}
                            - Number of parameters: 2
                            - Passes if the column value is less than the
                              smaller of the two arguments, or greater
                              than the larger of the two arguments.
                          - C{startswith}
                            - Number of parameters: 1
                            - Passes if the column value is a string which
                              starts with the parameter value.  The C{unicode}
                              function is called to transform the parameter
                              into a string unless it is None/null.  If the
                              parameter evaluates to an empty string or None,
                              no rows will pass.  Note that column values must
                              be actual strings to pass; the argument
                              C{['startswith', '123']} will match the string
                              '12345', but not the integer 12345.
                          - C{contains}
                            - Number of parameters: 1
                            - Passes if the column value is a string which
                              contains the parameter value.  The C{unicode}
                              function is called to transform the parameter
                              into a string unless it is None/null.  If the
                              parameter evaluates to an empty string or None,
                              no rows will pass.  Note that column values must
                              be actual strings to pass; the argument
                              C{['startswith', '123']} will match the string
                              '12345', but not the integer 12345. B{WARNING}:
                              Using this operator may prevent Kineta from using
                              an index to speed up the query.  The user is
                              cautioned against using C{contains} on very large
                              data sets.
                          - C{endswith}
                            - Number of parameters: 1
                            - Passes if the column value is a string which
                              ends with the parameter value.  The C{unicode}
                              function is called to transform the parameter
                              into a string unless it is None/null.  If the
                              parameter evaluates to an empty string or None,
                              no rows will pass.  Note that column values must
                              be actual strings to pass; the argument
                              C{['startswith', '123']} will match the string
                              '12345', but not the integer 12345. B{WARNING}:
                              Using this operator may prevent Kineta from using
                              an index to speed up the query.  The user is
                              cautioned against using C{endswith} on very large
                              data sets.

                      Comparisons are made using normal Python semantics,
                      and all values are well-ordered with respect to each
                      other.  When comparing one type with another, the
                      following order is maintained: None < list < number <
                      string.

        @type joins: list
        @param joins: A list of dictionaries, with each dict specifying
                      parameters for a join to the main table.  A join is a
                      scan of a table, the rows of which will be matched
                      against the parent table.  A single dict value will be
                      treated as a list with one item.  If None, no joins are
                      made.

                      Syntax::

                        joins = [dict(
                            table = "a table"|["a table", "an alias"],
                            on = {"parentcol1": "childcol1", ...},
                            fields = ["childcol1", ["subjoin"]],
                            where = {"childcol1": [operator, value1, ...],
                                     ...},
                            joins = dict(table="subjoin", ...),
                            outer = True|False,
                            order = [["childcol1", "asc"], ...]
                            ), ...]

                      All values default to None if they are not present,
                      unless otherwise noted.

                       - C{table}: The table to join, or a table name and a
                                   join alias.  If only a table is specified,
                                   the alias is taken to be the table name.  A
                                   table can be joined to multiple times in one
                                   query, and can join to itself, but all
                                   joins in the entire query must have a unique
                                   alias (including the main table), because
                                   the alias is used to distinguish columns in
                                   the result set.
                       - C{on}: A dict specifying the columns to join on.  Each
                                key in the dict is a column in the parent
                                table, and each corresponding value is a column
                                in the joined table.  All columns in the
                                key/value pairs must be equal to each other in
                                order for rows from the two tables to match
                                (i.e., Kineta currently only supports
                                I{equi-joins}).
                       - C{fields}: The columns to read from the joined table.
                                    These columns will be inserted in the
                                    parent's field list if the parent specifies
                                    C{[this_join_name]} in its fields.  If
                                    None, no fields will be inserted.  The
                                    syntax is the same as the I{fields}
                                    parameter to select; i.e., join fields can
                                    include the fields from their own child
                                    joins.
                       - C{where}: A set of conditions which specifies the rows
                                   in the joined table which are eligible to be
                                   matched against.  Has identical syntax to
                                   the main I{where} parameter.
                       - C{joins}: A list of sub-joins.  Each joined table can
                                   also be joined to any other number of
                                   tables, with the results processed
                                   recursively, with the same syntax applying
                                   as to the main parameter.
                       - C{outer}: A boolean indicating whether the join is
                                   "outer" or "inner".  The default if not
                                   specified is False, i.e. inner.  Inner means
                                   that a row is output whenever a row from the
                                   parent table matches a row in the joined
                                   table.  Outer produces a row whenever an
                                   inner join would, but also produces a row
                                   for each parent table row even if no matches
                                   are found for it, with the child fields in
                                   the result set to None.
                       - C{order}: A list indicating the order of rows returned
                                   within this join.  The syntax is identical
                                   to the main I{order} parameter.
        @type order: list
        @param order: A list containing columns and directions to order the
                      resulting rows by.

                      Syntax::

                        ["column_1", ["column_2", "asc"|"desc"|"a"|"d"], ...]

                      If a column is specified on its own, the rows will be
                      sorted by that column in ascending order.  Otherwise a
                      two-element list can be specified, the the second element
                      indicating the direction by either "asc" or "desc"
                      (these identifiers can be abbreviated "a" or "d" and are
                      not case-sensitive).

                      Columns refer to the main table only.  To order by a
                      column in a joined table, specify it in the order clause
                      for that join; the overall order of all the rows is
                      determined by the order the joins are specified in
                      (in left-to-right, depth-first order).

                      For example, to sort by t1.c1, t1.c2, and t2.c3, all in
                      descending order::

                        db.select(table="t1", order=[["c1", "d"], ["c2", "d"]],
                                  joins=dict(table="t2",
                                             order=[["c3", "d"]],
                                             ...)
                                  ...)

        @type limit: integer
        @param limit: Sets the maximum number of rows which will be returned
                      from the query.  After returning the number of rows
                      specified by this parameter, the query iterator will
                      complete immedietely.  Note that unless an order is
                      specified, the actual rows returned are likely to be
                      arbitrary.  If None, no limit is set the query will
                      return an unlimited number of rows.
        @type offset: integer
        @param offset: Sets the number of rows to skip before starting to
                       return rows from the query.  If this parameter
                       is larger than the number of rows available, no rows
                       will be returned.  Note that unless an order is
                       specified, the actual rows skipped are likely to be
                       arbitrary.  If None, no rows are skipped.
        @type txn: transaction
        @param txn: The transaction object to use.  If None, calls the method
                    outside the context of any transaction. Calling
                    select on a transaction object itself will
                    automatically call this method with the txn parameter set.
        @rtype: iterator
        @return: A reader object which can be iterated through to retrieve
                 rows which match the query.  In addition, the reader object
                 has the following methods/properties:

                  - C{field_names}: A property containing a list of two item
                                    lists, each of which contains a join/table
                                    name and a column name.

                  - C{as_dict()}: By default, rows are returned as lists of
                                  values.  Calling this method sets an internal
                                  flag which causes rows to be returned as
                                  two-level dictionaries.  For example, with
                                  C{field_names} equal to::

                                    [["t1", "c1"], ["t1", "c2"], ["t2", "c3"]]

                                  A reader that would normally return a row
                                  of::

                                    ["a", "b", "c"]

                                  would instead return::

                                    {"t1": {"c1": "a", "c2": "b"},
                                     "t2": {"c3": "c"}}

                                  after this method was called.  This method
                                  returns the reader object it was called on.

                  - close(): Closes the reader and releases resources
                             associated with it.  This method does not need to
                             be called twice if a new reader is created with
                             C{as_dict()}.  Callers must always ensure readers
                             are closed after use, however if they are created
                             as the initializer in a with: block, they will
                             automatically be closed and this method will not
                             need to be called.  Calling C{close()} on an
                             already closed reader has no effect.  Closing a
                             reader before iteration through all its rows has
                             completed is perfectly OK and will have no
                             ill-effects.
        """
        return self._select_or_explain(table, fields, where, joins, order,
                                       limit, offset, txn)

    def explain(self, table, fields, where=None, joins=None, order=None,
                limit=None, offset=None, after=False, txn=None):
        """
        Get a plan for executing a query, which is represented as a dictionary,
        possibly with other dictionaries embedded in it.  Each individual
        dictionary represents a logical sequence which Kineta will chain
        together to execute the query.

        The meanings of the parameters for this method are the same as for
        L{select}, with the exception of I{after}.

        For example, given the following call::

            a_plan = db.explain(table='a table', fields=['col1'],
                                order=['col1'])

        an example of a plan might be::

            {'strategy': 'sort', 'in_memory': True,
             'sub_plan':
                {'strategy': 'table_scan', 'table': 'a table'}}

        This indicates that Kineta would have performed a table scan on a table
        named 'a table', and fed the results to a separate sorting process.
        However, if 'a table' had an index defined which enabled Kineta to
        read the rows in sorted order, an alternate plan might be returned
        instead::

            {'strategy': 'index_scan', 'table': 'a table',
             'index': ['col1', 'a'], 'bookmark_lookup': False}

        @param table: See L{select}.
        @param fields: See L{select}.
        @param where: See L{select}.
        @param joins: See L{select}.
        @param order: See L{select}.
        @param limit: See L{select}.
        @param offset: See L{select}.
        @type after: boolean
        @param after: If True, Kineta will return the query plan only after
                      running the entire query internally. This means that the
                      plan will contain statistics on how many rows were
                      actually accessed and returned, but the explain call
                      could potentially take much longer.  The default of
                      False will only prepare the query as if it were to be
                      executed, and then return the plan immedietely.
        @type txn: transaction
        @param txn: See L{select}.
        @rtype: dict
        @return: A dict containing values which show how Kineta will or did
                 execute a query.
        """
        return self._select_or_explain(table, fields, where, joins, order,
                                       limit, offset, txn,
                                       explain=True, after=after)

    def update(self, table, values, where=None, limit=None, offset=None,
               order=None, txn=None):
        """
        Set existing rows in a table to a set of specified values.

        @type table: string
        @param table: The table to update.
        @type values: dict
        @param values: A dictionary, the keys of which are taken as column
                       names to update, and the values as the new values to
                       set the columns of matching rows to.  Values can be
                       numbers, strings, booleans, None, or lists containing
                       any of these four types.
        @type where: dict
        @param where: A set of constraints for which rows get updated. See the
                      L{select} method documentation for more details.
        @type limit: integer
        @param limit: Sets the maximum number of rows which will be updated.
        @type offset: integer
        @param offset: Sets the number of rows to skip before starting to
                       update.
        @type order: list
        @param order: The order to update rows in.  Only useful if the
                      I{limit} and/or I{offset} parameters are specified.
                      The syntax is identical to the I{order} parameter to
                      L{select}.
        @type txn: transaction
        @param txn: The transaction object to use.  If None, will run the
                    the method outside the context of any transaction. Calling
                    update on a transaction object itself will
                    automatically call this method with the txn parameter set.
        @rtype: integer
        @return: The number of rows updated.
        """
        if None in values:
            raise KinetaError("Cannot update a null column name")

        with _TxnLock(txn) as txnobj:
            tableid = self._db.dict.get_id(table, True)

            valuesbyid = {}
            for key, val in values.iteritems():
                valuesbyid[self._db.dict.get_id(key)] = query.update_value(val)

            return self._update_or_delete(tableid, where, limit, offset, order,
                                          valuesbyid, "update", txnobj)

    def delete(self, table, where=None, limit=None, offset=None, order=None,
               txn=None):
        """
        Delete existing rows from a table.

        @type table: string
        @param table: The table to delete from.
        @type where: dict
        @param where: A set of constraints for which rows get deleted. See the
                      L{select} method documentation for more details.
        @type limit: integer
        @param limit: Sets the maximum number of rows which will be deleted.
        @type offset: integer
        @param offset: Sets the number of rows to skip before starting to
                       delete.
        @type order: list
        @param order: The order to delete rows in.  Only useful if the
                      I{limit} and/or I{offset} parameters are specified.
                      The syntax is identical to the I{order} parameter to
                      L{select}.
        @type txn: transaction
        @param txn: The transaction object to use.  If None, will run the
                    the method outside the context of any transaction. Calling
                    delete on a transaction object itself will
                    automatically call this method with the txn parameter set.
        @rtype: integer
        @return: The number of rows deleted.
        """
        with _TxnLock(txn) as txnobj:
            tableid = self._db.dict.get_id(table, True)
            return self._update_or_delete(tableid, where, limit, offset,
                                          order, None, "delete", txnobj)

    def list_tables(self):
        """
        Lists the tables in the database.

        @rtype: list
        @return: A list of strings, which will be the names of the tables
                 currently in the database.  Tables are created when data is
                 inserted into them, and removed when the last row in them is
                 deleted.
        """
        ret = []
        with self._db.tables.lock() as lock:
            for tableid in self._db.tables.table_list(lock):
                ret.append(self._db.dict.get_name(tableid))
        return ret

    def list_fields(self, table):
        """
        Lists the fields in a table.  Since performing this operation
        requires reading through the entire table, calling this method is not
        recommended for production databases.

        @type table: string
        @param table: The table to list the fields from.
        @rtype: list
        @return: A list of strings, which will be the names of all the fields
                 which currently have values set in the table.  The fields are
                 returned in no particular order.
        """

        cols = set()
        tableid = self._db.dict.get_id(table, True)
        if tableid is not None:
            lock = self._db.tables.lock()
            try:
                strategy = query.get_field_list_strategy(self._db, None,
                                                         lock, tableid)
                try:
                    for row in strategy:
                        for k in row:
                            cols.add(self._db.dict.get_name(k))
                finally:
                    strategy.close()
            finally:
                lock.close()
        return list(cols)

    def list_indexes(self, table):
        """
        Lists the indexes on a table.

        @type table: string
        @param table: The table to list the indexes on.
        @rtype: list
        @return: A list of lists containing index specifications.  The format
                 of each sub-list i identical to that used for the I{order}
                 parameter to L{select}, except that each column/direction
                 pair is normalized (e.g. C{"a column"} becomes
                 C{["a column", "a"]})
        """
        ret = []
        tableid = self._db.dict.get_id(table, True)
        if tableid is not None:
            with self._db.tables.lock() as lock:
                indexes = self._db.tables.index_list(lock, tableid)
            if indexes is not None:
                for idx in indexes:
                    ret.append(utils.get_index_names(self._db, idx))
        return ret

    def create_index(self, table, columns, permanent=False):
        """
        Create an index.

        If the specified index already exists, no action is taken, except that
        an existing non-permanent index can be made permanent.

        The actual index creation takes place in a background thread, so
        users should not expect the index to exist unless L{wait_for_bg_tasks}
        is called after this method returns.

        @type table: string
        @param table: The table to create an index on.
        @type columns: list
        @param columns: The columns and directions to index.  The syntax is
                        identical to that used for the I{order} parameter to
                        L{select}.
        @type permanent: boolean
        @param permanent: If True, Kineta will not automatically remove the
                          index when a large amount of data has been written
                          without the index being utilized in a query.
        """
        tableid = self._db.dict.get_id(table, True)
        if tableid is None:
            return
        if not isinstance(columns, (tuple, list)):
            columns = [columns]
        ind = query.parse_index(self._db, columns)
        self._bgthread.task(lambda: self._db.tables.index_create(tableid,
                                                                 ind,
                                                                 permanent))

    def drop_index(self, table, columns):
        """
        Removes an index from the database.

        If the specified index does not exist, no action is taken.

        @type table: string
        @param table: The table to remove an index from.
        @type columns: list
        @param columns: A list of two-element lists which specify columns and
                        directions.  The format is identical to that used for
                        the I{order} parameter to L{select}.
        """
        tableid = self._db.dict.get_id(table, True)
        if tableid is not None:
            if not isinstance(columns, (tuple, list)):
                columns = [columns]
            ind = query.parse_index(self._db, columns)
            self._db.tables.index_drop(tableid, ind)

    def drop_table(self, table):
        """
        Removes a table from the database, including all of its data.
        Functionally, there is very little difference between deleting all the
        data from a table with L{delete} and dropping it, because Kineta allows
        inserting into or querying non-existent tables.  The main effects of
        using this method would be affecting the return value of L{list_tables}
        and resetting the I{identity} sequence number for the table.

        If the specified table does not exist, no action is taken.

        @type table: string
        @param table: The table to remove.
        """
        tableid = self._db.dict.get_id(table, True)
        if tableid is not None:
            num = self._db.tables.table_drop(tableid)

    def _select_or_explain(self, table, fields, where, joins, order,
                           limit, offset, txn, explain=False, after=False):
        limit, offset = query.parse_limit_offset(limit, offset)

        txnlock = _TxnLock(txn)
        try:
            opts, fieldmap = query.parse_queryoptions(self._db, table, fields,
                                                      where, joins, order,
                                                      self.limits,
                                                      (limit is not None and
                                                       limit == 0))
            lock = self._db.tables.lock()
            try:
                strategy = opts.get_strategy(txnlock.obj, lock)
                if explain:
                    try:
                        if after:
                            for row in strategy:
                                pass
                        return strategy.get_plan()
                    finally:
                        strategy.close()
                else:
                    return _SelectResult(fieldmap, lock, txnlock,
                                         self._bgthread, strategy,
                                         limit, offset)
            except: #pragma: no cover
                lock.close()
                raise
            finally:
                if explain:
                    lock.close()
        except: #pragma: no cover
            txnlock.release()
            raise
        else:
            if explain:
                txnlock.release()

    def _update_or_delete(self, tableid, where, limit, offset, order, values,
                          op, txnobj):
        whereobj = None
        if where is not None:
            whereobj = query.parse_where(self._db, where)

        limit, offset = query.parse_limit_offset(limit, offset)

        ordercols = None
        if order is not None:
            if not isinstance(order, (list, tuple)):
                order = [order]
            if len(order) > 0:
                coldict = dict(zip(map(str, fields), cols))
                ordercols = parse_order(db, order, coldict)

        if tableid is None:
            return 0

        opts = query.ScanOptions(self._db, tableid, None, set(), whereobj,
                                 ordercols, op, self.limits)
        num = 0
        with self._db.tables.lock() as lock:
            strategy = opts.get_strategy(txnobj, lock)
            try:
                for internalrow in strategy:
                    if offset is not None:
                        if num < offset:
                            num += 1
                            continue
                        else:
                            offset = None
                            num = 0
                    if limit is not None and num >= limit:
                        break
                    num += 1
                    row, rowid = internalrow.get_tuple(None)
                    if op == "update":
                        for key, val in values.iteritems():
                            if val is None and key in row:
                                del row[key]
                            elif val is not None:
                                row[key] = val
                        pickled = pickle.dumps(row, self._db.pickleproto)
                        strategy.update(rowid, pickled)
                    else:
                        strategy.delete()
                strategy.update_commit()
            finally:
                strategy.close()
        self._bgthread.task(lambda: self._db.tables.update_scores(num,
                                                                  tableid))
        return num

    def _insert_item(self, lock, tblobj, seqobj, columns, values,
                     hasidentity, txnobj):
        identity = None
        if hasidentity:
            identity = seqobj.get(1, txnobj)

        item = {}
        for i in xrange(len(columns)):
            val = values[i]
            if val is not None:
                item[self._db.dict.get_id(columns[i])] = \
                                       query.insert_value(val, identity)

        tblobj.append(pickle.dumps(item, self._db.pickleproto), txnobj)

        return identity

    @staticmethod
    def _check_insert_values(values):
        hasidentity = False
        for v in values:
            if isinstance(v, dict):
                if len(v) != 1 or not ("identity" in v):
                    raise KinetaError("Unknown custom value definition")
                i = v["identity"]
                if not isinstance(i, bool) or not i:
                    raise KinetaError("Identity value is %s, must be True"
                                                               % repr(i))
                hasidentity = True
        return hasidentity

_needs_txn = set(["begin", "insert", "insert_bulk",
                  "select", "update", "delete"])


class Rollback(Exception):
    """
    This exception type can be used to roll back transactions by raising it in
    a with: block where a transaction object is the current context.  The
    transaction object will swallow the exception and abort the current
    transaction::

        with db.begin() as txn:
            txn.insert("sometable", {"col1": "val1"})
            raise Rollback() # rolls back the insert statement and continues
                             # control after the with: block.
    """
    pass


class _Txn(utils.DeleteHelper):

    def _pre_init(self):
        self.txn = None

    def __init__(self, parent, obj):
        self.txn = obj
        self._parent = parent
        self.lock = threading.Lock()

    def commit(self):
        if self.txn is not None:
            try:
                self._parent._db.commit(self.txn)
            finally:
                self.txn = None

    def rollback(self):
        if self.txn is not None:
            try:
                self._parent._db.abort(self.txn)
            finally:
                self.txn = None

    def __del__(self):
        self.rollback()

    def __enter__(self):
        return self

    def __exit__(self, ex, val, tb):
        if tb is not None:
            self.rollback()
            return ex is Rollback
        else:
            self.commit()

    def __getattr__(self, name):
        attr = getattr(self.__dict__['_parent'], name)
        if name in _needs_txn:
            return lambda *args, **kwargs: attr(txn=self, *args, **kwargs)
        return attr


class _TxnLock(object):

    def __init__(self, txn):
        self._txn = txn
        self._acquire()

    @property
    def obj(self):
        if self._txn is not None:
            return self._txn.txn
        return None

    def _acquire(self):
        if self._txn is not None:
            self._txn.lock.acquire()

    def release(self):
        if self._txn is not None:
            self._txn.lock.release()

    def __enter__(self):
        return self.obj

    def __exit__(self, *exc_info):
        self.release()


class _SelectResult(utils.DeleteHelper):

    def _pre_init(self):
        self._strategy = None
        self._lock = None
        self._txnlock = None
        self._threadlock = None

    def __init__(self, fieldmap, lock, txnlock, bgthread, strategy,
                 limit, offset):
        self._fieldmap = fieldmap
        self.field_names = fieldmap.field_names
        self._lock = lock
        self._txnlock = txnlock
        self._strategy = strategy
        self._iterator = iter(strategy)
        self._bgthread = bgthread
        self._asdict = False
        self._limit = limit
        self._offset = offset
        self._cnt = 0
        # this is used by the server class only
        self._threadlock = threading.Lock()

    def __iter__(self):
        return self

    def __enter__(self):
        return self

    def __exit__(self, *exc_info):
        self.close()

    def __del__(self):
        self.close()

    def as_dict(self):
        self._asdict = True
        return self

    def next(self):
        try:
            if self._offset is not None:
                while self._cnt < self._offset:
                    self._iterator.next()
                    self._cnt += 1
                self._offset = None
                self._cnt = 0

            if self._limit is not None and self._cnt >= self._limit:
                raise StopIteration()
            self._cnt += 1

            r = self._fieldmap.get_select_value(self._iterator.next())
            if self._asdict:
                d = {}
                for n, v in zip(self.field_names, r):
                    d.setdefault(n[0], {})[n[1]] = v
                r = d
            return r
        except:
            self._bgthread.task(self._strategy.index_update)
            self.close()
            raise

    def close(self):
        if self._strategy is not None:
            self._strategy.close()
            self._strategy = None
        if self._lock is not None:
            self._lock.close()
            self._lock = None
        if self._txnlock is not None:
            self._txnlock.release()
            self._txnlock = None
        if self._threadlock is not None:
            self._threadlock = None
